/*
 * Copyright 2017 Kafdrop contributors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 *
 */

package kafdrop.controller;

import java.io.File;
import java.util.*;
import java.util.concurrent.*;

import javax.validation.Valid;
import javax.validation.constraints.Max;
import javax.validation.constraints.Min;
import javax.validation.constraints.NotNull;

import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import kafdrop.util.*;
import org.apache.commons.lang3.ArrayUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.validation.BindingResult;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.ModelAttribute;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;

import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;

import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiResponse;
import io.swagger.annotations.ApiResponses;
import kafdrop.config.MessageFormatConfiguration;
import kafdrop.config.MessageFormatConfiguration.MessageFormatProperties;
import kafdrop.config.ProtobufDescriptorConfiguration;
import kafdrop.config.ProtobufDescriptorConfiguration.ProtobufDescriptorProperties;
import kafdrop.config.SchemaRegistryConfiguration;
import kafdrop.config.SchemaRegistryConfiguration.SchemaRegistryProperties;
import kafdrop.model.MessageVO;
import kafdrop.model.TopicPartitionVO;
import kafdrop.model.TopicVO;
import kafdrop.service.KafkaMonitor;
import kafdrop.service.MessageInspector;
import kafdrop.service.TopicNotFoundException;

@Controller
public final class MessageController {
    private final KafkaMonitor kafkaMonitor;

    private final MessageInspector messageInspector;

    private final MessageFormatConfiguration.MessageFormatProperties messageFormatProperties;
    private final MessageFormatConfiguration.MessageFormatProperties keyFormatProperties;

    private final SchemaRegistryConfiguration.SchemaRegistryProperties schemaRegistryProperties;

    private final ProtobufDescriptorConfiguration.ProtobufDescriptorProperties protobufProperties;
    final ThreadFactory threadFactory = new ThreadFactoryBuilder()
            .setNameFormat("llww-filter-kafka-%d")
            .setDaemon(true)
            .build();
    private ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime()
            .availableProcessors(), threadFactory);

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    public MessageController(KafkaMonitor kafkaMonitor, MessageInspector messageInspector, MessageFormatProperties messageFormatProperties, MessageFormatProperties keyFormatProperties, SchemaRegistryProperties schemaRegistryProperties, ProtobufDescriptorProperties protobufProperties) {
        this.kafkaMonitor = kafkaMonitor;
        this.messageInspector = messageInspector;
        this.messageFormatProperties = messageFormatProperties;
        this.keyFormatProperties = keyFormatProperties;
        this.schemaRegistryProperties = schemaRegistryProperties;
        this.protobufProperties = protobufProperties;
    }

    /**
     * Human friendly view of reading all topic messages sorted by timestamp.
     *
     * @param topicName Name of topic
     * @param model
     * @return View for seeing all messages in a topic sorted by timestamp.
     */
    @GetMapping("/topic/{name:.+}/allmessages")
    public String viewAllMessages(@PathVariable("name") String topicName,
                                  Model model, @RequestParam(name = "count", required = false) Integer count) {
        final int size = (count != null ? count : 100);
        final MessageFormat defaultFormat = messageFormatProperties.getFormat();
        final MessageFormat defaultKeyFormat = keyFormatProperties.getFormat();
        final TopicVO topic = kafkaMonitor.getTopic(topicName)
                .orElseThrow(() -> new TopicNotFoundException(topicName));

        model.addAttribute("topic", topic);
        model.addAttribute("defaultFormat", defaultFormat);
        model.addAttribute("messageFormats", MessageFormat.values());
        model.addAttribute("keyFormats", KeyFormat.values());
        model.addAttribute("descFiles", protobufProperties.getDescFilesList());

        final var deserializers = new Deserializers(
                getDeserializer(topicName, defaultKeyFormat, "", ""),
                getDeserializer(topicName, defaultFormat, "", ""));

        final List<MessageVO> messages = messageInspector.getMessages(topicName, size, deserializers);

        for (TopicPartitionVO partition : topic.getPartitions()) {
            messages.addAll(messageInspector.getMessages(topicName,
                    partition.getId(),
                    partition.getFirstOffset(),
                    size,
                    deserializers));
        }

        messages.sort(Comparator.comparing(MessageVO::getTimestamp));
        model.addAttribute("messages", messages);

        return "topic-messages";
    }

    /**
     * Human friendly view of reading messages.
     *
     * @param topicName   Name of topic
     * @param messageForm Message form for submitting requests to view messages.
     * @param errors
     * @param model
     * @return View for seeing messages in a partition.
     */
    @GetMapping("/topic/{name:.+}/messages")
    public String viewMessageForm(@PathVariable("name") String topicName,
                                  @Valid @ModelAttribute("messageForm") PartitionOffsetInfo messageForm,
                                  BindingResult errors,
                                  Model model) {
        final long methodStart = System.currentTimeMillis();
        final MessageFormat defaultFormat = messageFormatProperties.getFormat();
        final MessageFormat defaultKeyFormat = keyFormatProperties.getFormat();

        if (messageForm.isEmpty()) {
            final PartitionOffsetInfo defaultForm = new PartitionOffsetInfo();

            defaultForm.setCount(100l);
            defaultForm.setOffset(0l);
            defaultForm.setPartition(0);
            defaultForm.setFormat(defaultFormat);
            defaultForm.setKeyFormat(defaultFormat);

            model.addAttribute("messageForm", defaultForm);
        }

        final TopicVO topic = kafkaMonitor.getTopic(topicName)
                .orElseThrow(() -> new TopicNotFoundException(topicName));
        model.addAttribute("topic", topic);

        model.addAttribute("defaultFormat", defaultFormat);
        model.addAttribute("messageFormats", MessageFormat.values());
        model.addAttribute("defaultKeyFormat", defaultKeyFormat);
        model.addAttribute("keyFormats", KeyFormat.values());
        model.addAttribute("descFiles", protobufProperties.getDescFilesList());

        if (!messageForm.isEmpty() && !errors.hasErrors()) {

            final var deserializers = new Deserializers(
                    getDeserializer(topicName, messageForm.getKeyFormat(), messageForm.getDescFile(), messageForm.getMsgTypeName()),
                    getDeserializer(topicName, messageForm.getFormat(), messageForm.getDescFile(), messageForm.getMsgTypeName())
            );
            final long start = System.currentTimeMillis();
            List<MessageVO> oriMessage = messageInspector.getMessages(topicName,
                    messageForm.getPartition(),
                    messageForm.getOffset(),
                    messageForm.getCount().intValue(),
                    deserializers);
            final long finish = System.currentTimeMillis();

            final String currentThreadName = Thread.currentThread().getName();
            logger.info("线程[{}],用时[{}]ms,获取[{}]条数据 ", currentThreadName, finish - start, oriMessage.size());

            final String filter = messageForm.getFilterCondition();
            if (!StringUtils.isEmpty(filter)) {
                final String[] split = filter.split(",");
                List<Split> splits = new ArrayList<>();
                if (!ArrayUtils.isEmpty(split)) {
                    for (String s : split) {
                        final String[] split1 = s.split(":");
                        final Split split2 = new Split();
                        split2.setKey(split1[0]);
                        if (split1.length == 2) {
                            split2.setValue(split1[1]);
                        }
                        splits.add(split2);
                    }
                }
                if (!CollectionUtils.isEmpty(splits)) {
                    final int size = oriMessage.size();
                    if (size <= 100) {
                        final LinkedList<MessageVO> messageVOS = getMessageVOS(oriMessage, splits);
                        oriMessage = messageVOS;
                    } else {
                        CopyOnWriteArrayList<MessageVO> copyOnWriteArrayList = new CopyOnWriteArrayList<>();
                        final int availableProcessors = Runtime.getRuntime().availableProcessors();
                        CountDownLatch cdl = new CountDownLatch(availableProcessors);
                        final int fac = size / availableProcessors;
                        for (int i = 0; i < availableProcessors; i++) {
                            List<MessageVO> messageVOS1;
                            final int fromIndex = i * fac;
                            if (i == availableProcessors - 1) {
                                messageVOS1 = oriMessage.subList(fromIndex, size);

                            } else {
                                final int end = fromIndex + fac;
                                messageVOS1 = oriMessage.subList(fromIndex, end);
                            }

                            List<MessageVO> finalMessageVOS = messageVOS1;
                            executorService.submit(() -> {
                                try {
                                    final long aStart = System.currentTimeMillis();
                                    final String thredName = Thread.currentThread().getName();
                                    final LinkedList<MessageVO> messageVOS = getMessageVOS(finalMessageVOS, splits);
                                    copyOnWriteArrayList.addAll(messageVOS);
                                    cdl.countDown();
                                    final long cost = System.currentTimeMillis() - aStart;
                                    final int oriSize = finalMessageVOS.size();
                                    final int filterdSize = messageVOS.size();
                                    logger.info("线程[{}] 用时[{}]ms,过滤[{}]条数据,共[{}]条数据匹配，规则为[{}]", thredName, cost, oriSize, filterdSize, splits);

                                } catch (Throwable e) {
                                    logger.error(e.getMessage());
                                }

                            });


                        }
                        try {
                            cdl.await(100000, TimeUnit.SECONDS);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        oriMessage = copyOnWriteArrayList;

                    }

                }
            }


            model.addAttribute("messages",
                    oriMessage);

        }
        logger.info("获取消息处理完成 cost(ms):[{}] topic:[{}] form:[{}] ",System.currentTimeMillis()-methodStart,topicName,messageForm);
        return "message-inspector";
    }

    private LinkedList<MessageVO> getMessageVOS(List<MessageVO> messages, List<Split> splits) {
        final int size = messages.size();
        final LinkedList<MessageVO> messageVOS = new LinkedList<>();
        for (int i = 0; i < size; i++) {
            MessageVO messageVO = messages.get(i);
            final String message = messageVO.getMessage();
            final JsonNode jsonNode = JsonUtil.jsonNodeOf(message);
            boolean flag = true;
            for (Split tempSplit : splits) {

                final String key = Optional.ofNullable(tempSplit.getKey()).map(String::trim).orElse(null);
                if (StringUtils.isEmpty(key)) {
                    break;
                }
                final String value = Optional.ofNullable(tempSplit.getValue()).map(String::trim).orElse(null);

                if (!StringUtils.isEmpty(value)) {
                    final String asText = Optional
                            .ofNullable(jsonNode.get(key))
                            .map(JsonNode::textValue)
                            .orElse("");
                    if (!asText.contains(value)) {
                        flag = false;
                        break;
                    }
                } else {
                    if (!message.contains(key)) {
                        flag = false;
                        break;
                    }
                }


            }
            if (flag) {
                messageVOS.add(messageVO);
            }
        }
        logger.debug("getMessageVOS count [{}]", messageVOS.size());
        return messageVOS;
    }

    private static class Split {
        private String key;
        private String value;

        public String getKey() {
            return key;
        }

        public void setKey(String key) {
            this.key = key;
        }

        public String getValue() {
            return value;
        }

        public void setValue(String value) {
            this.value = value;
        }

        @Override
        public String toString() {
            return "Split{" +
                    "key='" + key + '\'' +
                    ", value='" + value + '\'' +
                    '}';
        }
    }

    /**
     * Returns the selected nessagr format based on the
     * form submission
     *
     * @param format String representation of format name
     * @return
     */
    private MessageFormat getSelectedMessageFormat(String format) {
        if ("AVRO".equalsIgnoreCase(format)) {
            return MessageFormat.AVRO;
        } else if ("PROTOBUF".equalsIgnoreCase(format)) {
            return MessageFormat.PROTOBUF;
        } else if ("MSGPACK".equalsIgnoreCase(format)) {
            return MessageFormat.MSGPACK;
        } else {
            return MessageFormat.DEFAULT;
        }
    }


    /**
     * Return a JSON list of all partition offset info for the given topic. If specific partition
     * and offset parameters are given, then this returns actual kafka messages from that partition
     * (if the offsets are valid; if invalid offsets are passed then the message list is empty).
     *
     * @param topicName Name of topic.
     * @return Offset or message data.
     */
    @ApiOperation(value = "getPartitionOrMessages", notes = "Get offset or message data for a topic. Without query params returns all partitions with offset data. With query params, returns actual messages (if valid offsets are provided).")
    @ApiResponses(value = {
            @ApiResponse(code = 200, message = "Success", response = List.class),
            @ApiResponse(code = 404, message = "Invalid topic name")
    })
    @RequestMapping(method = RequestMethod.GET, value = "/topic/{name:.+}/messages", produces = MediaType.APPLICATION_JSON_VALUE)
    public @ResponseBody
    List<Object> getPartitionOrMessages(
            @PathVariable("name") String topicName,
            @RequestParam(name = "partition", required = false) Integer partition,
            @RequestParam(name = "offset", required = false) Long offset,
            @RequestParam(name = "count", required = false) Integer count,
            @RequestParam(name = "format", required = false) String format,
            @RequestParam(name = "keyFormat", required = false) String keyFormat,
            @RequestParam(name = "descFile", required = false) String descFile,
            @RequestParam(name = "msgTypeName", required = false) String msgTypeName
    ) {
        if (partition == null || offset == null || count == null) {
            final TopicVO topic = kafkaMonitor.getTopic(topicName)
                    .orElseThrow(() -> new TopicNotFoundException(topicName));

            List<Object> partitionList = new ArrayList<>();
            topic.getPartitions().forEach(vo -> partitionList.add(new PartitionOffsetInfo(vo.getId(), vo.getFirstOffset(), vo.getSize())));

            return partitionList;
        } else {

            final var deserializers = new Deserializers(
                    getDeserializer(topicName, getSelectedMessageFormat(keyFormat), descFile, msgTypeName),
                    getDeserializer(topicName, getSelectedMessageFormat(format), descFile, msgTypeName));

            List<Object> messages = new ArrayList<>();
            List<MessageVO> vos = messageInspector.getMessages(
                    topicName,
                    partition,
                    offset,
                    count,
                    deserializers);

            if (vos != null) {
                messages.addAll(vos);
            }

            return messages;
        }
    }

    private MessageDeserializer getDeserializer(String topicName, MessageFormat format, String descFile, String msgTypeName) {
        final MessageDeserializer deserializer;

        if (format == MessageFormat.AVRO) {
            final var schemaRegistryUrl = schemaRegistryProperties.getConnect();
            final var schemaRegistryAuth = schemaRegistryProperties.getAuth();

            deserializer = new AvroMessageDeserializer(topicName, schemaRegistryUrl, schemaRegistryAuth);
        } else if (format == MessageFormat.PROTOBUF) {
            // filter the input file name
            final var descFileName = descFile.replace(".desc", "")
                    .replaceAll("\\.", "")
                    .replaceAll("/", "");
            final var fullDescFile = protobufProperties.getDirectory() + File.separator + descFileName + ".desc";
            deserializer = new ProtobufMessageDeserializer(topicName, fullDescFile, msgTypeName);
        } else if (format == MessageFormat.MSGPACK) {
            deserializer = new MsgPackMessageDeserializer();
        } else {
            deserializer = new DefaultMessageDeserializer();
        }

        return deserializer;
    }

    /**
     * Encapsulates offset data for a single partition.
     */
    public static class PartitionOffsetInfo {
        @NotNull
        @Min(0)
        private Integer partition;

        /**
         * Need to clean this up. We're re-using this form for the JSON message API
         * and it's a bit confusing to have the Java variable and JSON field named
         * differently.
         */
        @NotNull
        @Min(0)
        @JsonProperty("firstOffset")
        private Long offset;

        /**
         * Need to clean this up. We're re-using this form for the JSON message API
         * and it's a bit confusing to have the Java variable and JSON field named
         * differently.
         */
        @NotNull
        @Min(1)
        @Max(Long.MAX_VALUE)
        @JsonProperty("lastOffset")
        private Long count;


        private String filterCondition;

        private MessageFormat format;

        private MessageFormat keyFormat;

        private String descFile;

        private String msgTypeName;

        public PartitionOffsetInfo(int partition, long offset, long count, MessageFormat format) {
            this.partition = partition;
            this.offset = offset;
            this.count = count;
            this.format = format;
        }

        public PartitionOffsetInfo(int partition, long offset, long count) {
            this(partition, offset, count, MessageFormat.DEFAULT);
        }

        public PartitionOffsetInfo() {

        }

        @JsonIgnore
        public boolean isEmpty() {
            return partition == null && offset == null && (count == null || count == 1);
        }

        public Integer getPartition() {
            return partition;
        }

        public void setPartition(Integer partition) {
            this.partition = partition;
        }

        public Long getOffset() {
            return offset;
        }

        public void setOffset(Long offset) {
            this.offset = offset;
        }

        public Long getCount() {
            return count;
        }

        public void setCount(Long count) {
            this.count = count;
        }

        public MessageFormat getKeyFormat() {
            return keyFormat;
        }

        public void setKeyFormat(MessageFormat keyFormat) {
            this.keyFormat = keyFormat;
        }

        public MessageFormat getFormat() {
            return format;
        }

        public void setFormat(MessageFormat format) {
            this.format = format;
        }

        public String getDescFile() {
            return descFile;
        }

        public void setDescFile(String descFile) {
            this.descFile = descFile;
        }

        public String getMsgTypeName() {
            return msgTypeName;
        }

        public void setMsgTypeName(String msgTypeName) {
            this.msgTypeName = msgTypeName;
        }

        public String getFilterCondition() {
            return filterCondition;
        }

        public void setFilterCondition(String filterCondition) {
            this.filterCondition = filterCondition;
        }

        @Override
        public String toString() {
            return "PartitionOffsetInfo{" +
                    "partition=" + partition +
                    ", offset=" + offset +
                    ", count=" + count +
                    ", filterCondition='" + filterCondition + '\'' +
                    ", format=" + format +
                    ", keyFormat=" + keyFormat +
                    ", descFile='" + descFile + '\'' +
                    ", msgTypeName='" + msgTypeName + '\'' +
                    '}';
        }
    }
}
